package com.gis507.mapreduce.AISClassify;

import com.gis507.mapreduce.AISBean.AisClassifyBean;
import com.gis507.mapreduce.Utils.UtilsClass;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;

/**
 * outPut K:AISClassifyBean  V:NullWritable
 */
public class wy_AISClassReducer extends Reducer<Text, AisClassifyBean,AisClassifyBean, NullWritable> {

    private static double Sampling =50; //设置采样
    private static int Segment = 5; //分段距离
    private static int PointMinNum=20; //估计点个数最小
    private static int StopSpeed = 2; //停留速度
    private static int FilterSpeed = 15; //速度过滤

    @Override
    protected void reduce(Text key, Iterable<AisClassifyBean> values, Context context) throws IOException, InterruptedException {
        ArrayList<AisClassifyBean> data = new ArrayList<>();
        for (AisClassifyBean value : values) {
            data.add(new AisClassifyBean(value.getMmsi(),value.getShipType(),value.getNacStatusCN()
                    ,value.getDraught(),value.getHeading(),value.getCourse(),value.getSpeed(),value.getDest(),
                    value.getUnixTime(),value.getLon_d(),value.getLat_d(),value.getSeaRange()));
            sort(data);
        }
        ArrayList<AisClassifyBean> list = new ArrayList<>();
        ArrayList<AisClassifyBean> list_list = new ArrayList<>();
        int i=0, j=1;
        double dis=0.0;
        while(j<data.size()){
            if (data.get(j).getUnixTime()-data.get(i).getUnixTime()>=Sampling)
                if (data.get(j).getSpeed()<=FilterSpeed) {
                    list.add(new AisClassifyBean(data.get(j).getMmsi(), data.get(j).getShipType(),
                            data.get(j).getNacStatusCN(), data.get(j).getDraught(),data.get(j).getHeading()
                    ,data.get(j).getCourse(),data.get(j).getSpeed(),data.get(j).getDest(),data.get(j).getUnixTime()
                    ,data.get(j).getLon_d(),data.get(j).getLat_d(),data.get(j).getSeaRange()));
                    j++;
                }
                else j++;
            else j++;
        }
        j=0;
        while (j<list.size()-2){
            if (list.get(j).getSpeed()>StopSpeed) {
                list_list.add(new AisClassifyBean(list.get(j).getMmsi(), list.get(j).getShipType(),
                        list.get(j).getNacStatusCN(), list.get(j).getDraught(),list.get(j).getHeading()
                        ,list.get(j).getCourse(),list.get(j).getSpeed(),list.get(j).getDest(),list.get(j).getUnixTime()
                        ,list.get(j).getLon_d(),list.get(j).getLat_d(),list.get(j).getSeaRange()));
                j++;
            }
            else {
                j++;
                while (j<list.size()-2){
                    if (list.get(j).getSpeed()>StopSpeed){
                        list_list.add(new AisClassifyBean(list.get(j).getMmsi(), list.get(j).getShipType(),
                                list.get(j).getNacStatusCN(), list.get(j).getDraught(),list.get(j).getHeading()
                                ,list.get(j).getCourse(),list.get(j).getSpeed(),list.get(j).getDest(),list.get(j).getUnixTime()
                                ,list.get(j).getLon_d(),list.get(j).getLat_d(),list.get(j).getSeaRange()));
                        break;
                    }
                    else j++;
                }
                i=j+1;
                dis=GetDistance(list.get(i).getLat_d(),list.get(i).getLon_d(),list.get(j).getLat_d(),list.get(j).getLon_d());
                if (dis<=Segment)
                    list_list.add(new AisClassifyBean(list.get(i).getMmsi(), list.get(i).getShipType(),
                            list.get(i).getNacStatusCN(), list.get(i).getDraught(),list.get(i).getHeading()
                            ,list.get(i).getCourse(),list.get(i).getSpeed(),list.get(i).getDest(),list.get(i).getUnixTime()
                            ,list.get(i).getLon_d(),list.get(i).getLat_d(),list.get(i).getSeaRange()));
                else if (list_list.size()<PointMinNum)
                    list_list.clear();
                else {
                    int m=1,InsertNum;
                    while (m<list_list.size()){
                        InsertNum = (int) (GetDistance(list.get(m).getLat_d(),list.get(m).getLon_d(),list.get(m-1).getLat_d(),list.get(m-1).getLon_d())/0.6+0.5);
                        if(InsertNum>0){
                            for (int n=1; n<=InsertNum;n++){
                                double InsertLat = list_list.get(m-1).getLat_d() + list_list.get(m).getLat_d()-list_list.get(m-1).getLat_d()*n/(InsertNum+1);
                                double InsertLon = list_list.get(m-1).getLon_d() + list_list.get(m).getLon_d()-list_list.get(m-1).getLon_d()*n/(InsertNum+1);
                                long InsertUnixTime =(long) (list_list.get(m-1).getUnixTime() + list_list.get(m).getUnixTime()-list_list.get(m-1).getUnixTime()*n/(InsertNum+1));
                                double InsertSpeed = list_list.get(m-1).getSpeed() + list_list.get(m).getSpeed()-list_list.get(m-1).getSpeed()*n/(InsertNum+1);
                                list.add(new AisClassifyBean(list_list.get(m-1).getMmsi(),list_list.get(m-1).getShipType(),list_list.get(m-1).getNacStatusCN()
                                        ,list_list.get(m-1).getDraught(),list_list.get(m-1).getHeading(),list_list.get(m-1).getCourse(),InsertSpeed,
                                        list_list.get(m-1).getDest(),InsertUnixTime,InsertLon,InsertLat,list_list.get(m-1).getSeaRange()));
                            }
                            m++;
                        }
                        else m++;
                    }
                }
            }
        }
        ArrayList<AisClassifyBean> out = new ArrayList<>();
        for (AisClassifyBean value : list) {
            out.add(new AisClassifyBean(value.getMmsi(),value.getShipType(),value.getNacStatusCN()
                    ,value.getDraught(),value.getHeading(),value.getCourse(),value.getSpeed(),value.getDest(),
                    value.getUnixTime(),value.getLon_d(),value.getLat_d(),value.getSeaRange()));
            sort(out);
        }
//        System.out.println("保留点数：                      "+m);
        for (AisClassifyBean l : out) context.write(l,NullWritable.get());
    }

    public static void sort( ArrayList<AisClassifyBean> a) {
        for(int i=a.size()-1;i > 0;i--)
        {
            if(a.get(i).getUnixTime() < a.get(i-1).getUnixTime())
            {
                AisClassifyBean temp = a.get(i-1);
                a.set(i-1,a.get(i));
                a.set(i,temp);
            }
        }
    }

    private final static double EARTH_RADIUS = 6378.137;//地球半径
    private static double rad(double d) {
        return d * Math.PI / 180.0;
    }
    public static double GetDistance(double lat1, double lng1, double lat2, double lng2) {
        double radLat1 = rad(lat1);
        double radLat2 = rad(lat2);
        double a = radLat1 - radLat2;
        double b = rad(lng1) - rad(lng2);
        double s = 2 * Math.asin(Math.sqrt(Math.pow(Math.sin(a / 2), 2) +
                Math.cos(radLat1) * Math.cos(radLat2) * Math.pow(Math.sin(b / 2), 2)));
        s = s * EARTH_RADIUS;
        s = new BigDecimal(s).setScale(3, BigDecimal.ROUND_HALF_UP).doubleValue();
        return s;
    }

}
